package com.cskaoyan.mall.promo.mq.producer;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import com.cskaoyan.mall.promo.mq.properties.CreatePromoOrderProducerProperties;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName CreatePromoOrderProducer
 * @Description TODO
 * @Author XiaoWJ
 * @Date 2022/9/29 20:54
 **/
@Component
public class CreatePromoOrderProducer {

    private TransactionMQProducer producer;

    @Autowired
    CreatePromoOrderProducerProperties properties;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient client;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer(properties.getId());

        producer.setNamesrvAddr(properties.getAddr());

        producer.setTransactionListener(new TransactionListener() {

            // 执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                // 扣减秒杀商品库存
                Map<String,Long> paramMap = (Map<String, Long>) o;
                Long productId = paramMap.get("productId");
                Long psId = paramMap.get("psId");

                // 指定保存事务执行状态的key
                String key = "local_promo_transaction" + message.getTransactionId();

                Integer affectRows = 0;

                // 锁的id需要唯一
                String lockKey = "local_promo_transaction" + "-" + productId + "-" + psId;
                RLock rLock = client.getLock(lockKey);

                try {
                    rLock.lock();
                    affectRows = promoItemMapper.decreaseStock(productId, psId);
                }finally {
                    // 释放锁
                    rLock.unlock();
                }

                // 扣减成功，保存本地事务的执行状态
                if(affectRows > 0) {
                    cacheManager.setCache(key, "COMMIT_MESSAGE",1);
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                // 扣减失败，保存本地事务的执行状态
                cacheManager.setCache(key, "ROLLBACK_MESSAGE",1);
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            // 检查本地事务的执行状态
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 查redis缓存
                String key = "local_promo_transaction" + messageExt.getMsgId();

                String checkCache = cacheManager.checkCache(key);
                if("COMMIT_MESSAGE".equals(checkCache)) {
                    return LocalTransactionState.COMMIT_MESSAGE;

                }else if("ROLLBACK_MESSAGE".equals(checkCache)) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    public boolean sendMessageCreateOrder(CreateSeckillOrderRequest seckillOrderRequest,
                                          Long productId, Long psId) {

        // 新建消息对象
        Message message = new Message();

        // 设置主题
        message.setTopic(properties.getTopic());

        // 设置消息
        String messageBody = JSON.toJSONString(seckillOrderRequest);

        message.setBody(messageBody.getBytes(Charset.forName(properties.getCharset())));

        // 构造一个map, 为本地事务的执行传入参数
        HashMap<String, Long> paramMap = new HashMap<>();
        paramMap.put("productId",productId);
        paramMap.put("psId",psId);

        // 发送消息
        TransactionSendResult transactionSendResult = null;
        try {
            transactionSendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        // 消息发送的结果不为并且响应状态为已提交，消息发送成功
        if(transactionSendResult != null && LocalTransactionState.COMMIT_MESSAGE
                .equals(transactionSendResult.getLocalTransactionState())) {
            return true;

        }
        return false;
    }

}
